package com.danan.realtime.app.dwd;

import com.danan.realtime.app.BaseSqlApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.util.SinkUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/1/25
 */
public class DwdApp_6_trade_order_refund_detail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdApp_6_trade_order_refund_detail().init(4006,3,"DwdApp_6_trade_order_refund_success_detail");
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 读取ods_db数据
        readOdsDb(tableEnv,"DwdApp_6_trade_order_refund_success_detail");
        // 2 过滤退单表数据
        filterRefundedOrderInfo(tableEnv);
        // 3 过滤订单数据
        filterOrderInfo(tableEnv);
        // 4 读取字典表
        readBaseDic(tableEnv);
        // 5 join
        join(tableEnv);
        // 6 将结果写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table target (" +
                "id string," +
                "user_id string," +
                "order_id string," +
                "sku_id string," +
                "province_id string," +
                "date_id string," +
                "create_time string," +
                "refund_type_code string," +
                "refund_type_name string," +
                "refund_reason_type_code string," +
                "refund_reason_type_name string," +
                "refund_reason_txt string," +
                "refund_num string," +
                "refund_amount string," +
                "ts bigint " +
                ")" +
                SinkUtil.getKafkaWith(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void join(StreamTableEnvironment tableEnv) {
        // 设置数据的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(7 * 24 * 60 * 60 + 5));
        // join
        Table joinResult = tableEnv.sqlQuery("select " +
                "ro.id," +
                "ro.user_id," +
                "ro.order_id," +
                "ro.sku_id," +
                "oi.province_id," +
                "date_format(ro.create_time,'yyyy-MM-dd') date_id," +
                "ro.create_time," +
                "ro.refund_type," +
                "dic1.dic_name," +
                "ro.refund_reason_type," +
                "dic2.dic_name," +
                "ro.refund_reason_txt," +
                "ro.refund_num," +
                "ro.refund_amount," +
                "ro.ts " +
                "from order_refund_info ro " +
                "join order_info oi on ro.order_id = oi.id " +
                "join base_dic for system_time as of ro.pt as dic1 on ro.refund_type = dic1.dic_code " +
                "join base_dic for system_time as of ro.pt as dic2 on ro.refund_reason_type = dic2.dic_code");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void filterOrderInfo(StreamTableEnvironment tableEnv) {
        Table orderInfo = tableEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['province_id'] province_id " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_info' " +
                "and `type`='update' " +
                "and `old`['order_status'] is not null " +
                "and `data`['order_status']='1005'");
        tableEnv.createTemporaryView("order_info",orderInfo);
    }

    private void filterRefundedOrderInfo(StreamTableEnvironment tableEnv) {
        Table orderRefundInfo = tableEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['order_id'] order_id," +
                "data['sku_id'] sku_id," +
                "data['refund_type'] refund_type," +
                "data['refund_num'] refund_num," +
                "data['refund_amount'] refund_amount," +
                "data['refund_reason_type'] refund_reason_type," +
                "data['refund_reason_txt'] refund_reason_txt," +
                "data['create_time'] create_time," +
                "pt," +
                "ts " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_refund_info' " +
                "and `type`='insert'");
        tableEnv.createTemporaryView("order_refund_info",orderRefundInfo);
    }
}
